package com.smsc.headend.task.engine.handler.impl.collection;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.enums.errorcode.TaskHandlerErrorCode;
import com.smsc.headend.common.utils.UDateUtils;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.data.dto.DataItemDTO;
import com.smsc.headend.module.data.dto.LpQueryDTO;
import com.smsc.headend.module.data.dto.TaskDataDTO;
import com.smsc.headend.module.data.dto.TaskPacketDTO;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.data.enums.CollectionSource;
import com.smsc.headend.module.protocol.DeviceData;
import com.smsc.headend.module.protocol.cosem.bo.CosemData;
import com.smsc.headend.module.report.entity.LogScheduleTaskDevice;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.dto.TaskReadData;
import com.smsc.headend.module.task.dto.TaskResult;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISTaskHandlerException;
import com.smsc.headend.task.engine.handler.TaskHandler;
import com.smsc.headend.task.engine.handler.impl.collection.strategy.CollectionStrategy;
import com.smsc.headend.task.engine.handler.impl.collection.strategy.CollectionStrategyFactory;
import com.smsc.headend.task.engine.service.CollMeterTaskPlanService;
import com.smsc.headend.task.engine.service.DataProcessService;
import com.smsc.headend.task.engine.service.DeviceStatusService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.calc.RdEnergyDailyService;
import com.smsc.headend.task.engine.service.task.LogScheduleTaskDeviceService;
import com.smsc.headend.task.engine.service.task.impl.TaskServiceImpl;
import com.smsc.headend.task.engine.utils.TaskParamUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.*;

@Slf4j
public class ReadLoadProfileBufferHandler implements TaskHandler {
    public static final String RECOLLECT_TASK_SEQ = "recollectTaskSeq";
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    DataProcessService dataProcessService;
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    TaskServiceImpl taskService;
    @Autowired
    AssetService assetService;
    @Autowired
    CollectionStrategyFactory collectionStrategyFactory;
    @Autowired
    LogScheduleTaskDeviceService logScheduleTaskDeviceService;
    @Autowired
    DeviceStatusService deviceStatusService;
    @Autowired
    RdEnergyDailyService rdEnergyDailyService;
    @Autowired
    CollMeterTaskPlanService collMeterTaskPlanService;

    @Override
    public int handle(Task task, AtomicTask atomicTask, DeviceData deviceData, String actionResult) throws UDISTaskHandlerException {
        if (!(deviceData instanceof CosemData)) {
            log.error("result={}, data={}", actionResult, deviceData);
            throw new UDISTaskHandlerException(TaskHandlerErrorCode.DEVICE_DATA_TYPE_NOT_SUPPORT.toString());
        }
        CosemData cosemData = (CosemData) deviceData;
        Map<String, Object> param = getParameterMap(task, atomicTask);
        Long lpTemplateId = MapUtil.getLong(param, "lpTemplateId");
        Long lpDataItemId = MapUtil.getLong(param, "dataItemId");
        if (!param.containsKey(RECOLLECT_TASK_SEQ) && lpTemplateId == null) {
            throw new UDISTaskHandlerException(TaskHandlerErrorCode.NO_LP_TEMPLATE_ID.toString());
        }
        List<DataItemDTO> dataItemCosemList = dataProcessService.getLoadProfileCaptureObjects(lpTemplateId, lpDataItemId, task.getMeterId());
        if (CollectionUtil.isEmpty(dataItemCosemList)) {
            throw new UDISTaskHandlerException(TaskHandlerErrorCode.INVALID_CONFIG_CAP_OBJECTS.toString());
        }
        log.debug("capture objects {}", JSONUtil.toJsonStr(dataItemCosemList));
        Meter meter = assetService.getMeterById(task.getMeterId());
        Boolean newVersionFlag = isNewDLMSGatewayData(cosemData);
        if (newVersionFlag) {
            //目前DLMS Gateway只支持单块电表的数据采集。即 Array长度只为1
            CosemData firstArrayElement = cosemData.getChildAt(0);
            String meterSnInDLMSGateway = firstArrayElement.getChildAt(0).getValue();
            //暂时不做profile的obis校验
//            String obis = firstArrayElement.getChildAt(1).getValue();
            if (!meter.getMfgSn().equalsIgnoreCase(meterSnInDLMSGateway)) {
                throw new UDISTaskHandlerException(TaskHandlerErrorCode.DLMS_GATEWAY_METER_SN_NOT_MATCH.toString());
            }
            cosemData = firstArrayElement.getChildAt(2);
        }
        CollectionSource collectionSource = getCollectionType(task, param);
        try {
            List<RdMeterReadsLp> meterReadsCol = new ArrayList<>();
            if (CollUtil.isNotEmpty(cosemData.getChild())) {
                meterReadsCol = dataProcessService.getRdMeterReadLpList(cosemData, lpTemplateId, lpDataItemId, dataItemCosemList, meter, collectionSource);
            }

            CollectionStrategy collectionStrategy = collectionStrategyFactory.getCollectionStrategy(collectionSource);
            //data filter
            List<RdMeterReadsLp> filteredRdMeterReadsLp = collectionStrategy.filterCollectionData(task, atomicTask, param, meterReadsCol);
            //save profile data
            TaskReadData readData = new TaskReadData();
            readData.setProfileId(lpDataItemId);
            readData.setTaskNo(task.getTaskNo());
            readData.setIsODR(task.getIsODR());
            if (CollUtil.isNotEmpty(filteredRdMeterReadsLp)) {
                deviceStatusService.updateMeterReadDataStatus(meter);
                readData.setTaskMeterReads(filteredRdMeterReadsLp);
                dataProcessService.saveTaskReadData(readData, collectionSource);
                if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
                    logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Success.getCode(), LogScheduleTaskDevice.StatusDescEnum.CollectedSuccess.getCode());
                }
                rdEnergyDailyService.calcDailyEnergyInterval(readData);
            } else {
                if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
                    logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Failed.getCode(), LogScheduleTaskDevice.StatusDescEnum.EmptyData.getCode());
                }
            }
            //callback
            int result = collectionStrategy.callback(meter, readData, task, atomicTask, param, dataItemCosemList, filteredRdMeterReadsLp);
            return result;
        } catch (Exception e) {
            if (task.getIsODR()) {
                if (param.containsKey(RECOLLECT_TASK_SEQ)) {
                    Map<String, Object> data = MapUtil.newHashMap();
                    data.put("errorText", "Unexpected error");
                    taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), param.get(RECOLLECT_TASK_SEQ).toString(), data, Long.valueOf(1 * 60 * 60), false);
                } else {
                    taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), lpDataItemId.toString(), "Unexpected error", Long.valueOf(1 * 60 * 60), false);
                }
            }
            log.error("Unexpected error", e);
        }
        return -1;
    }

    /**
     * OldVersionDLMSGateway Or Meter COSEM profile DataBuffer Example
     * <GetResponse>
     * ....\
     * <Data>
     * <Array Qty="01" >
     * <Structure Qty="04" >
     * <!--2022-04-05 22:15:00-->
     * <OctetString Value="07E6040502160F00008000FF" />
     * <UInt8 Value="04" />
     * <UInt32 Value="00000005" />
     * <UInt32 Value="00000006" />
     * </Structure>
     * </Array>
     * </Data>
     * ....
     * </GetResponse>
     * <p>
     * DLMSGateway new version
     * <GetResponse>
     * ....\
     * <Data>
     * <Array Qty="01" >
     * <Structure Qty="03" >
     * <String Value="240000000001" />
     * <!--1.0.98.1.1.255-->
     * <OctetString Value="0100620101FF" />
     * <Array Qty="02" >
     * <Structure Qty="04" >
     * <!--2022-03-22 02:00:00-->
     * <OctetString Value="07E6031602020000FF800000" />
     * <UInt8 Value="00" />
     * <UInt32 Value="00EAA89D" />
     * <UInt32 Value="00000000" />
     * </Structure>
     * </Array>
     * </Structure>
     * </Array>
     * </Data>
     * ....
     * </GetResponse>
     *
     * @param cosemData
     * @return
     */
    private Boolean isNewDLMSGatewayData(CosemData cosemData) {
        if (!"Array".equalsIgnoreCase(cosemData.getType())) {
            return false;
        }
        if (cosemData.getChild() == null) {
            return false;
        }
        CosemData firstArrayElement = cosemData.getChildAt(0);
        if (firstArrayElement.getChild() == null || firstArrayElement.getChild().size() != 3) {
            return false;
        }
        //Structure 中第三个元素为Array类型 则为新版本DLMS Gateway的协议
        if ("Array".equalsIgnoreCase(firstArrayElement.getChildAt(2).getType())) {
            return true;
        }
        return false;
    }

    private CollectionSource getCollectionType(Task task, Map<String, Object> param) {
        if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
            return CollectionSource.SCH;
        }

        if (TaskType.AutoReCollectProfile.toString().equalsIgnoreCase(task.getTaskType())) {
            return CollectionSource.ARR;
        }

        if (TaskType.UAA_MeterReads.toString().equalsIgnoreCase(task.getTaskType()) || TaskType.ODR_Profile.toString().equalsIgnoreCase(task.getTaskType()) || TaskType.Interface_ODR_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
            if (param.containsKey(RECOLLECT_TASK_SEQ)) {
                return CollectionSource.MARR;
            }
            return CollectionSource.ODR;
        }
        return null;
    }

    @Override
    public int handleFailure(Task task, AtomicTask atomicTask, String message) {
        log.info("error msg:{}", message);
        if (task.getIsODR()) {
            Map<String, Object> param = getParameterMap(task, atomicTask);
            Long lpDataItemId = Long.valueOf(String.valueOf(param.get("dataItemId")));
            // key 回曲线的dataItemId给web
            taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), Convert.toStr(lpDataItemId), message, 3600l, false);
        }
        Map<String, Object> param = getParameterMap(task, atomicTask);
        String compensateQuery = MapUtil.getStr(param, "compensateQuery");
        boolean isVendingMonthLyTask = false;
        if (StrUtil.equalsIgnoreCase("daily", compensateQuery)) {
            isVendingMonthLyTask = true;
        }
        if (isVendingMonthLyTask) {
            handVendingFailedMonthData(task, param);
            return -1;
        }

        if (StrUtil.equalsIgnoreCase(task.getTaskType(), TaskType.Interface_ODR_Profile.name()) || StrUtil.equalsIgnoreCase(task.getTaskType(), TaskType.UAA_MeterReads.name())) {
            TaskResult taskResult = new TaskResult();
            taskResult.setMessageId(task.getMessageId());
            taskResult.setTaskNo(task.getTaskNo());
            taskResult.setExceptionCause(message);
            Map<String, Object> dataItemMap = getParameterMap(task, atomicTask);
            Map<String, Object> taskResultMap = JSONUtil.toBean(task.getParamJson(), Map.class);
            taskResultMap.putAll(dataItemMap);
            getParameterMap(task, atomicTask);
            Long lpDataItemId = Long.valueOf(String.valueOf(dataItemMap.get("dataItemId")));
            Meter meter = assetService.getMeterById(task.getMeterId());
            List<TaskPacketDTO> taskPacketDTOList = taskService.getTaskPacket(task.getTaskNo());
            if (CollectionUtil.isNotEmpty(taskPacketDTOList)) {
                taskResultMap.put("packets", taskPacketDTOList);
            }
            taskResultMap.put("meterModelId", meter.getModelId());
            taskResultMap.put("meterCustomSn", meter.getCustomerSn());
            taskResultMap.put("profileDataItemId", lpDataItemId);
            taskResult.setParamJson(JSONUtil.toJsonStr(taskResultMap));
            kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(Charset.forName("utf-8")));
        }
        if (StrUtil.equalsIgnoreCase(task.getTaskType(), TaskType.ScheduleRead_Profile.name())) {
            TaskDataDTO latestTv = Optional.ofNullable(taskDAO.popAllTaskData(task.getTaskNo())).orElse(Collections.emptyList()).stream()
                    .filter(dto -> dto.getKey().equals("latestTv")).findAny().orElse(null);
            if (latestTv != null) {
                Long dataItemId = MapUtil.getLong(TaskParamUtils.getParameterMap(task, atomicTask), "dataItemId");
                log.warn("updateProgress:meterId:{},lpId:{}", task.getMeterId(), dataItemId);
                collMeterTaskPlanService.saveMeterCollectProgress(Convert.toLong(latestTv.getValue()), task.getMeterId(), dataItemId);
            }
            logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Failed.getCode(), LogScheduleTaskDevice.StatusDescEnum.DecodeDataFailed.getCode());
        }
        return -1;
    }

    private void handVendingFailedMonthData(Task task, Map<String, Object> param ) {
        try {
            //如果失败，查daily数据进行返回或计算
            String profileStartTime = MapUtil.getStr(param, "profileStartTime");
            Long lpTv = UDateUtils.SECONDS_OF_ONE_HOUR + Long.parseLong(profileStartTime);

            LpQueryDTO lpQueryDTO = new LpQueryDTO();
            lpQueryDTO.setMessageId(task.getMessageId());
            lpQueryDTO.setMeterId(task.getMeterId());
            lpQueryDTO.setLpDataItemId(1020070001L);
            lpQueryDTO.setTv(lpTv);

            List<RdMeterReadsLp> lpData = dataProcessService.queryElectricMeterDataByDaily(lpQueryDTO);

            TaskResult taskResult = new TaskResult();
            taskResult.setMessageId(lpQueryDTO.getMessageId());
            taskResult.setResultData(lpData);
            Map<String, Object> taskResultMap = new HashMap<>();
            Meter meter = assetService.getMeterById(lpQueryDTO.getMeterId());
            taskResultMap.put("meterModelId", meter.getModelId());
            taskResultMap.put("meterCustomSn", meter.getCustomerSn());
            taskResultMap.put("profileDataItemId", lpQueryDTO.getLpDataItemId().longValue());
            taskResult.setParamJson(JSONUtil.toJsonStr(taskResultMap));
            kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            log.error("Exception ", e);
        }

    }


}

